use std::time::Duration;

use axum::extract::ws::Message;
use axum::response::IntoResponse;
use axum::Form;
use axum::{
    extract::{ws::WebSocket, Query, State, WebSocketUpgrade},
    Json,
};
use futures::{SinkExt, StreamExt};
use serde::Deserialize;
use serde_json::{json, Value};

use crate::stores::sql;
use crate::{misc::utils, misc::wx, services, stores::redisx, SharedState};

// vtype code类型，1 为登录code，需要从微信后台获取openid ；2为session code，在redis中保存了用户code，有效期为24h。
#[derive(Debug, Deserialize)]
pub struct VCodeParam {
    pub vcode: String,
    pub vtype: i32,
}
pub async fn get_vcode_1109(
    State(state): State<SharedState>,
    Query(params): Query<VCodeParam>,
) -> Json<Value> {
    let vcode = params.vcode;
    let vtype = params.vtype;
    match vtype {
        1 => {
            // 根据code获取openid。
            let cf = &state.conf.mp;
            let openid = wx::get_openid_bycode(&cf.app_id, &cf.app_secret, &vcode)
                .await
                .unwrap_or_else(|e| {
                    tracing::error!("get_vcode_1109 err, {}", e);
                    "".to_string()
                });
            if openid == "" {
                return Json(json!({"code":-1,"msg":"参数错误"}));
            }
            let pool = &state.db;
            let cnt = services::user::get_user_count(&pool, openid.clone())
                .await
                .unwrap();
            if cnt == 0 {
                let rand_str = utils::get_rand_str(6);
                let nickname = format!("mu{}", rand_str.to_lowercase());
                services::user::create_user(&pool, openid.clone(), nickname, 1)
                    .await
                    .unwrap();
            }
            let rs = &state.rs;
            let sessionid = utils::get_timestamp() + 86400;
            let token_key = format!("token1109-{}", sessionid);
            if let Err(e) = redisx::set_ex(rs, token_key, openid.clone(), 86400).await {
                tracing::error!("redis setex error, {} ", e);
                return Json(json!({"code":-1,"msg":"系统错误"}));
            }

            let code = utils::get_rand_num(6);
            let rcode = format!("vcode1109-{}", code);
            if let Err(e) = redisx::set_ex(rs, rcode, openid.clone(), 300).await {
                tracing::error!("redis setex error, {} ", e);
                return Json(json!({"code":-1,"msg":"系统错误"}));
            }

            return Json(
                json!({"code":1,"msg":"获取成功","data":{"rcode":code,"sessionid":sessionid}}),
            );
        }
        2 => {
            let rs = &state.rs;
            let token_key = format!("token1109-{}", vcode);
            let openid = redisx::get(rs, &token_key).await.unwrap_or("".to_string());
            if openid == "" {
                return Json(json!({"code":-1,"msg":"token无效"}));
            }

            let code = utils::get_rand_num(6);
            let rcode = format!("vcode1109-{}", code);
            if let Err(e) = redisx::set_ex(rs, rcode, openid.clone(), 300).await {
                tracing::error!("redis setex error, {} ", e);
                return Json(json!({"code":-1,"msg":"系统错误"}));
            }

            return Json(json!({"code":1,"msg":"获取成功","data":{"rcode":code}}));
        }
        _ => return Json(json!({"code":-1,"msg":"参数不正确"})),
    }
}

#[derive(Deserialize)]
pub struct LoginParam {
    pub vcode: String,
}
pub async fn login_1109(
    State(state): State<SharedState>,
    Form(params): Form<LoginParam>,
) -> Json<Value> {
    let vcode = params.vcode;
    let rs = &state.rs;
    let vcode_key = format!("vcode1109-{}", vcode);
    let openid = redisx::get(rs, &vcode_key).await.unwrap_or("".to_string());
    if openid == "" {
        tracing::debug!("{} is invalid", vcode);
        return Json(json!({"code":-1,"msg":"token无效"}));
    }
    let _ = redisx::del(rs, &vcode_key).await;
    let db = &state.db;
    if let Ok(user) = sql::query_user(&db, openid.clone()).await {
        return Json(
            json!({"code":1,"msg":"成功","data":{"openid":user.openid,"nickname":user.nickname}}),
        );
    }
    Json(json!({"code":-1,"msg":"参数错误"}))
}

#[derive(Deserialize)]
pub struct WsConnectParam {
    pub openid: String,
}

pub async fn ws_connect_1111(
    ws: WebSocketUpgrade,
    State(state): State<SharedState>,
    Query(params): Query<WsConnectParam>,
) -> impl IntoResponse {
    let openid = params.openid;
    let conf = &state.conf;
    let content = format!("用户使用聊天室小工具了，{}", openid);
    let _ = utils::send_mp_notify(&conf.mp.robot_url, content).await;
    ws.on_upgrade(|socket| websocket(socket, state, openid))
}

async fn websocket(stream: WebSocket, state: SharedState, openid: String) {
    let db = &state.db;
    let user = sql::query_user(&db, openid.clone()).await.unwrap();
    let name = user.nickname;
    // By splitting, we can send and receive at the same time.
    tracing::debug!("{} open websocket", name);

    // By splitting, we can send and receive at the same time.
    let (mut sender, mut receiver) = stream.split();
    let mut interval = tokio::time::interval(Duration::from_secs(55));
    let mut rx = state.ws.tx.subscribe();

    // Now send the "joined" message to all subscribers.
    let msg = format!("{name} 加入房间");
    tracing::debug!("{msg}");
    let _ = state.ws.tx.send(msg);

    let tx = state.ws.tx.clone();
    let name1 = name.clone();

    let ws_task = tokio::spawn(async move {
        loop {
            tokio::select! {
                rmsg = rx.recv() => {
                    match rmsg {
                        Ok(m) => {
                            if sender.send(Message::Text(m)).await.is_err(){
                                break;
                            }
                        },
                        Err(e) => {
                            tracing::error!("get message err,{}",e);
                            break;
                        },
                    }
                }
                _ = interval.tick() => {
                    if sender.send(Message::Ping(vec![])).await.is_err() {
                        tracing::debug!("websocket ping error");
                        break;
                    }
                }

                smsg=receiver.next()=>{
                    match smsg{
                        Some(Ok(Message::Text(text))) =>{
                            let _ = tx.send(format!("{name1}-> {text}"));
                        }
                        Some(Ok(Message::Close(_)))=>{
                            tracing::debug!("receive close message");
                            break;
                        }
                        // Some(Ok(Message::Pong(_)))=>{
                        //     tracing::debug!("receive pong message");
                        // }
                        Some(Err(e))=> {
                            tracing::debug!("the err is {:?}",e);
                        }
                        Some(_)=>{}
                        None => break,
                    }
                }
            }
        }
    });

    let _ = ws_task.await;
    let name2 = name.clone();

    // Send "user left" message (similar to "joined" above).
    let msg = format!("{name2} 离开房间");
    tracing::debug!("{msg}");
    let _ = state.ws.tx.send(msg);

    // Remove username from map so new clients can take it again.
    // state.ws.peer_map.lock().unwrap().remove(&name2);
}

#[derive(Deserialize)]
pub struct UpdateNickNameParam {
    pub openid: String,
    pub nickname: String,
}

pub async fn update_nickname_1111(
    State(state): State<SharedState>,
    Form(params): Form<UpdateNickNameParam>,
) -> Json<Value> {
    let openid = params.openid;
    let nickname = params.nickname;
    let db = &state.db;
    if let Err(e) = sql::update_user_nickname(&db, openid, nickname).await {
        tracing::error!("update_nickname err, {}", e);
        return Json(json!({"code":-1,"msg":"参数错误"}));
    }
    Json(json!({"code":1,"msg":"修改成功"}))
}
